package cn._51doit.flink.day08;

import com.alibaba.fastjson.JSON;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

/**
 * ProcessFunction是Flink底层的流式计算的算子，可以方法Flink底层的属性和方法
 * <p>
 * ProcessFunction的三中功能：
 * 1.一条一条的处理数据
 * 2.使用State（值针对于KeyedStream）
 * 3.使用定时器（值针对于KeyedStream）
 */

public class ProcessFunctionDemo1 {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //{"name": "laozhao", "age": 18, "fv": 9999.99}
        //{"name": "laoduan", "age": 18, "fv": 99.99
        DataStreamSource<String> lines = env.socketTextStream("localhost", 8888);

        SingleOutputStreamOperator<Boy> res = lines.process(new ProcessFunction<String, Boy>() {

            @Override
            public void processElement(String line, Context ctx, Collector<Boy> out) throws Exception {

                try {
                    Boy boy = JSON.parseObject(line, Boy.class);
                    out.collect(boy);
                } catch (Exception e) {
                    //e.printStackTrace();
                }

            }
        });

        res.print();

        env.execute();

    }


    public static class Boy {
        public String name;
        public Integer age;
        public Double fv;


        @Override
        public String toString() {
            return "Boy{" +
                    "name='" + name + '\'' +
                    ", age=" + age +
                    ", fv=" + fv +
                    '}';
        }
    }
}
